package org.gokeep.communication.demo.group;

import com.google.protobuf.ByteString;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.gokeep.common.util.DateUtil;
import org.gokeep.communication.demo.TestData;
import org.gokeep.communication.message.MessageBuilder;
import org.gokeep.communication.proto.ChatProto;
import org.gokeep.communication.proto.ChatRequestProto;
import org.gokeep.communication.proto.CommandProto;
import org.gokeep.communication.proto.MessageProto;
import org.smartboot.socket.Protocol;
import org.smartboot.socket.StateMachineEnum;
import org.smartboot.socket.extension.processor.AbstractMessageProcessor;
import org.smartboot.socket.extension.protocol.FixedLengthBytesProtocol;
import org.smartboot.socket.transport.AioQuickClient;
import org.smartboot.socket.transport.AioSession;
import org.smartboot.socket.transport.WriteBuffer;

import java.io.IOException;
import java.nio.channels.AsynchronousChannelGroup;

@Slf4j
public class MessageClient {

    public static MessageClient builder() {
        return new MessageClient();
    }

    /**
     * 测试初始化建立连接
     */
    public AioSession start() {
        AioQuickClient aioQuickClient = new AioQuickClient("localhost", 56781, protocol(), processor());
        aioQuickClient.setReadBufferSize(2048);
        try {
            return aioQuickClient.start(AsynchronousChannelGroup.withFixedThreadPool(1, Thread::new));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 定义消息协议
     * 把ByteArray转换为Message对象
     * @return
     */
    private Protocol<MessageProto.Message> protocol() {
        return new FixedLengthBytesProtocol<>() {
            @Override
            @SneakyThrows
            protected MessageProto.Message decode(byte[] bytes, AioSession aioSession) {
                return MessageProto.Message.parseFrom(bytes);
            }
        };
    }

    /**
     * 消息处理
     * @return
     */
    public AbstractMessageProcessor<MessageProto.Message> processor() {
        return new AbstractMessageProcessor<>() {
            @SneakyThrows
            @Override
            public void process0(AioSession session, MessageProto.Message message) {
                if (message.getCommand().equals(CommandProto.CommandEnum.HEART)) {
                    MessageProto.Message heartAckMessage = message.toBuilder()
                            .setCommand(CommandProto.CommandEnum.HEART_ACK)
                            .build();
                    WriteBuffer writeBuffer = session.writeBuffer();
                    byte[] bytes = heartAckMessage.toByteArray();
                    writeBuffer.writeInt(bytes.length);
                    writeBuffer.write(bytes);
                    writeBuffer.flush();
                    log.info("HeartACK: client -> server, sessionId: [{}], number: [{}]", session.getSessionID(),message.getSerializeNo());
                    return;
                }
                //log.info("Client receive command: {} ", message.getCommand());
                ByteString byteString = message.getData();
                ChatRequestProto.ChatRequest clientChatMessage = ChatRequestProto.ChatRequest.parseFrom(byteString);
                ChatProto.Chat chat = clientChatMessage.getChat();
                String groupId = chat.getGroupId();
                String fromNickName = TestData.getUserById(chat.getFrom()).getNickname();
                if (StringUtils.isBlank(groupId)) {
                    System.out.printf("%s（%s） -> %s%n", fromNickName, DateUtil.timeToStr(System.currentTimeMillis()), chat.getContent());
                }else {
                    System.out.printf("%s[groupId:%s]（%s） -> %s%n", fromNickName, groupId,  DateUtil.timeToStr(System.currentTimeMillis()), chat.getContent());
                }
                System.out.print("回复:");
            }

            @Override
            public void stateEvent0(AioSession session, StateMachineEnum stateMachineEnum, Throwable throwable) {
                log.info("与服务器建立通信连接，状态: {}", stateMachineEnum, throwable);
                if (throwable != null) {
                    throwable.printStackTrace();
                }
            }
        };
    }

    /**
     * 发送消息
     * @param message
     */
    public void send(AioSession session, MessageProto.Message message) {
        send(session, message, 1, 0);
    }

    @SneakyThrows
    public void send(AioSession session, MessageProto.Message message, int count, long onceSleep) {
        WriteBuffer writeBuffer = session.writeBuffer();
        byte[] bytes = message.toByteArray();
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < count; i++) {
            writeBuffer.writeInt(bytes.length);
            writeBuffer.write(bytes);
            writeBuffer.flush();
            if (onceSleep != 0L) {
                Thread.sleep(onceSleep);
            }
        }
        long endTime = System.currentTimeMillis();
        //log.debug("{}: Send message use time is : {} ms", new DateTime().toTimeStr(), endTime - startTime);
    }

    /**
     * 建立连接
     * @param session
     */
    public void connection(AioSession session, String token) {
        MessageProto.Message.Builder messageBuilder = MessageBuilder.builder().initMessageBuilder(CommandProto.CommandEnum.CONNECTION);
        messageBuilder.setToken(token);
        send(session, messageBuilder.build());
    }
}
